/**
 * Software License, Version 1.0
 * 
 * Copyright 2003 The Trustees of Indiana University.  All rights reserved.
 * 
 *
 *Redistribution and use in source and binary forms, with or without 
 *modification, are permitted provided that the following conditions are met:
 *
 *1) All redistributions of source code must retain the above copyright notice,
 * the list of authors in the original source code, this list of conditions and
 * the disclaimer listed in this license;
 *2) All redistributions in binary form must reproduce the above copyright 
 * notice, this list of conditions and the disclaimer listed in this license in
 * the documentation and/or other materials provided with the distribution;
 *3) Any documentation included with all redistributions must include the 
 * following acknowledgement:
 *
 *"This product includes software developed by the Community Grids Lab. For 
 * further information contact the Community Grids Lab at 
 * http://communitygrids.iu.edu/."
 *
 * Alternatively, this acknowledgement may appear in the software itself, and 
 * wherever such third-party acknowledgments normally appear.
 * 
 *4) The name Indiana University or Community Grids Lab or NaradaBrokering, 
 * shall not be used to endorse or promote products derived from this software 
 * without prior written permission from Indiana University.  For written 
 * permission, please contact the Advanced Research and Technology Institute 
 * ("ARTI") at 351 West 10th Street, Indianapolis, Indiana 46202.
 *5) Products derived from this software may not be called NaradaBrokering, 
 * nor may Indiana University or Community Grids Lab or NaradaBrokering appear
 * in their name, without prior written permission of ARTI.
 * 
 *
 * Indiana University provides no reassurances that the source code provided 
 * does not infringe the patent or any other intellectual property rights of 
 * any other entity.  Indiana University disclaims any liability to any 
 * recipient for claims brought by any other entity based on infringement of 
 * intellectual property rights or otherwise.  
 *
 *LICENSEE UNDERSTANDS THAT SOFTWARE IS PROVIDED "AS IS" FOR WHICH NO 
 *WARRANTIES AS TO CAPABILITIES OR ACCURACY ARE MADE. INDIANA UNIVERSITY GIVES
 *NO WARRANTIES AND MAKES NO REPRESENTATION THAT SOFTWARE IS FREE OF 
 *INFRINGEMENT OF THIRD PARTY PATENT, COPYRIGHT, OR OTHER PROPRIETARY RIGHTS. 
 *INDIANA UNIVERSITY MAKES NO WARRANTIES THAT SOFTWARE IS FREE FROM "BUGS", 
 *"VIRUSES", "TROJAN HORSES", "TRAP DOORS", "WORMS", OR OTHER HARMFUL CODE.  
 *LICENSEE ASSUMES THE ENTIRE RISK AS TO THE PERFORMANCE OF SOFTWARE AND/OR 
 *ASSOCIATED MATERIALS, AND TO THE PERFORMANCE AND VALIDITY OF INFORMATION 
 *GENERATED USING SOFTWARE.
 */
package cgl.narada.node;

import java.io.BufferedReader;
import java.io.BufferedWriter;
import java.io.File;
import java.io.FileReader;
import java.io.FileWriter;
import java.io.IOException;
import java.io.InputStreamReader;
import java.util.Properties;
import java.util.StringTokenizer;

import cgl.narada.protocol.ClientProtocolHandler;
import cgl.narada.protocol.Protocol;
import cgl.narada.transport.TransportException;

/** This is the client node for our class. This class is responsible for
    <ul><li> Publishing events
    <li> Expressing interest in certain event - their subscription predicates
    <li> Being able to <i>roam </i> in response to node failures and so on.

    @author Shrideep Pallickara
    $Revision$
    $Date$
*/

public class ClientNode implements NodeDebugFlags {
  /** The Client protocol Handler */
  private ClientProtocolHandler clientProtocolHandler;
  
  /** A buffered writer for gathering testing statistics */
  private BufferedWriter bw, bw2, bw3, bw4 /*for jitter*/;
  
  /** the publish rate ... testing purposes only */
  private int pubRate =1;
  private int numOfMessages = 1;
  /** This is used to fine tune our publish rates */
  private int messagesAtATime = 1;
  private int matchRate=1;
  private int matchingMessages;

  private int selector=1;
  private int messageSize;
  private int tracker=0;
  
  private int[] messageSizes = {30, 50, 100, 150, 200, 250, 300, 350, 400, 450, 500, 1000, 1500, 2000, 2500, 3000, 3500, 4000, 4500, 5000};
  
  private int[] matchRates = {100};
  private int[] publishRates = {1};
  private int[] numMessageCompensation = {1};
  private int[] virtualPublishRates = {1};
  //private int[] messageSizes = {30, 50, 75, 100, 125, 150, 175, 200, 225, 250, 300, 325, 350, 375, 400, 425, 450};
  //private int[] matchRates = {100, 50, 33, 25, 20, 15, 10, 5, 4, 3, 2, 1};
  //private int[] matchRates = {15, 10, 5, 4, 3, 2, 1};
  //private int[] publishRates = { 20, 50, 100, 150, 200, 
  //				 250, 300, 350, 400, 450, 
  //				 500, 550, 600, 650, 700, 
  //				 750, 800, 850, 900, 1000};
  //private int[] numMessageCompensation = { 1 /*20*/,  
  //					   1 /* 50 */, 2 /*100 */, 3 /*150*/,
  //					   4 /* 200 */, 5 /*250 */ , 6 /*300*/,
  //					   7 /* 350 */, 8 /*400 */, 9 /*450*/,
  //					   10 /* 500 */, 11 /*550*/,
  //					   12 /* 600 */, 13 /*650*/,
  //					   14 /* 700 */, 15 /*750*/,
  //					   16 /* 800 */, 17 /*850*/,
  //                                       18 /* 900 */,  20 /** 1000 */};
  
  //private int[] virtualPublishRates 
  //= { 25 /*20*/ , 100 /* 50 */, 
  //    100 /*100 */, 100 /*150*/,
  //    100 /* 200 */, 100 /*250 */ , 
  //    100 /* 300 */, 100 /*350 */ , 
  //    100 /* 400 */, 100 /*450 */ , 
  //    100 /* 500 */, 100 /*550 */ , 
  //    100 /* 600 */, 100 /*650 */ , 
  //    100 /* 700 */, 100 /*750 */,
  //    100 /* 800 */, 100 /*850 */,
  //    100 /* 900 */, 100 /** 1000 */}; 
  
  private String resultsDirectory= "results";
  float pubValueFromFile=0;
  public ClientNode() {
    clientProtocolHandler = new ClientProtocolHandler();
    clientProtocolHandler.setClientNode(this);
  }
  
  /** This method sets the publish rate ... testing purposes only 
      @param pubRate The publish rate
  */
  public void setRates(int pubRate, int numOfMessages,
		       int messagesAtTime) {
    this.pubRate = pubRate;
    this.numOfMessages = numOfMessages;
    this.messagesAtATime = messagesAtTime;
  }
  
  public void setPublishRate(int pRate) {
    clientProtocolHandler.setPublishRate(pRate);
  }
  
  public void setMatchRate(int matchRate) {
    this.matchRate  = 100/matchRate;
    matchingMessages = matchRate;
  }
  
  
  public int[] getMessageSizes() {
    return messageSizes;
  }

  public int[] getPublishRates() {
    return publishRates;
  }

  public int[] getMatchRates() {
    return matchRates;
  }
  /** This method serves as an intermediary between the transport layer
      and protocol layer and is responsible for setting up the server node.
  */
  public void setUpNode(int portNum, byte transportType) {
    Properties props = new Properties();
    props.put("TCPServerPort", Integer.toString(portNum));
    
    try {
      clientProtocolHandler.loadCommunicationsOfType(props, "tcp");
    } catch (TransportException transEx) {
      System.out.println(transEx);
    }
  }

    /** This method serves as an intermediary between the transport layer
      and protocol layer and is responsible for setting up the server node.
  */
  public void setUpNodeNew(Properties props, String transportType) {
    try {
    clientProtocolHandler.loadCommunicationsOfType(props, transportType);
    } catch (TransportException transEx) {
      System.out.println(transEx);
    }
  }

  /** Initiates a connection with the server that posseses the information
      contained within the request */
  public void connectTo(String hostName, int portNum, byte transport) {
    Properties props = new Properties();
    props.put("hostname", hostName);
    props.put("portnum", Integer.toString(portNum));
    //props.put("host", hostName);
    //props.put("port", Integer.toString(portNum));
    try {
      String _linkId = clientProtocolHandler.setupLink(props, "tcp");
      System.out.println(_linkId);
    } catch (TransportException transEx) {
      System.out.println(transEx);
    }
  }
  /** This method is reponsible for issuing a node address request to the
      server in question. 
      @param server The server to which this request is being issued.
  */
  public void nodeAddressRequest(String server) {
    clientProtocolHandler.nodeAddressRequest(server);
  }
  
  /** Propagates the interest in a subscription to the server in question.
      This method cannot be issued unitil the client has been assigned a 
      logical address by the system.
      @param server The server to which this client is attached.
      @param subscription The subscription predicate for the client 
  */
  public void profilePropagationRequest(String server, String subscription) {
    clientProtocolHandler.profilePropagationRequest(server,subscription);
  }
  
  /** This method is responsible for publishing an event to the system.
      @param server The server to which this request is being issued.
      @param event The event that is being published 
      @param minSize Minimum Size of the event being published.*/
  public void eventPropagationRequest(String server, String event, 
				      int minSize) {
    int messagesWithMatchingClient=0;
    long timeNow = System.currentTimeMillis();
    int sleepTime = 1000/pubRate;
    selector = 0;
    for (int i=0; i < numOfMessages; ) {
      try {
	//Thread.sleep(2);
	Thread.sleep(sleepTime);
      } catch (InterruptedException ie) {
	ie.printStackTrace();
      }
      for (int j=0; j < messagesAtATime; j++) {
	tracker++;
	selector++;
	if (selector > matchRate && 
	    messagesWithMatchingClient < matchingMessages) {
	  selector = 1;
	}
	if (selector == 1) {
	  messagesWithMatchingClient++;
	}
	String eventString = "A=a,B=b,C=" +selector + ",MesgNbr=" + tracker;
	clientProtocolHandler.eventPropagationRequest(server, 
					      eventString, minSize);
	//clientProtocolHandler.eventPropagationRequest(server, 
	//					      event, minSize);
	i++;
      } /*messagesAtATime */ 
    } /*numOfMessages */
    System.out.println("Messages with matching client ="  +
		       messagesWithMatchingClient);
    long publishTime = System.currentTimeMillis() -timeNow;
    double publishRate = ((double)numOfMessages/(double)publishTime)*1000;
    System.out.println("Publish rate = "  + (float)publishRate 
		       + "messages/Sec" + "\n");
    /** This value would then be read by the client */
    writePublishRate((float)publishRate);
  }
  
  /** This method is responsible for publishing a PERSISTENT event to the 
      system.
      @param server The server to which this request is being issued.
      @param event The event that is being published 
      @param minSize Minimum Size of the event being published.*/
  public void persistentEventPropagationRequest(String server, String event,
						int minSize) {
    int messagesWithMatchingClient=0;
    long timeNow = System.currentTimeMillis();
    int sleepTime = 1000/pubRate;
    selector = 0;
    for (int i=0; i < numOfMessages; ) {
      try {
	//Thread.sleep(2);
	Thread.sleep(sleepTime);
      } catch (InterruptedException ie) {
	ie.printStackTrace();
      }
      for (int j=0; j < messagesAtATime; j++) {
	tracker++;
	selector++;
	if (selector > matchRate && 
	    messagesWithMatchingClient < matchingMessages) {
	  selector = 1;
	}
	if (selector == 1) {
	  messagesWithMatchingClient++;
	}
	String eventString = "A=a,B=b,C=" +selector + ",MesgNbr=" + tracker;
	clientProtocolHandler.persistentEventPropagationRequest(server, 
								eventString, 
								minSize);
	i++;
      } /*messagesAtATime */ 
    } /*numOfMessages */
    System.out.println("Messages with matching client ="  +
		       messagesWithMatchingClient);
    long publishTime = System.currentTimeMillis() -timeNow;
    double publishRate = ((double)numOfMessages/(double)publishTime)*1000;
    System.out.println("Publish rate = "  + (float)publishRate 
		       + "messages/Sec" + "\n");
    /** This value would then be read by the client */
    writePublishRate((float)publishRate);
  }
  
    
  public void gatherStats(int _pubRate, int _matchRate, int _messageSize) {
    try {
      String fileName = "Pub" + _pubRate + "Match" + _matchRate
	+ "Size" + _messageSize +  ".txt";
      bw = new BufferedWriter(new FileWriter(new File(resultsDirectory,
						      fileName)));
      System.out.println("Buffered Reader " + fileName + "is ready ...");
      short[] latencies = clientProtocolHandler.getLatencyArray();
      
      for (int i=0; i < latencies.length; i++) {
	String text = latencies[i] + " ";
	bw.write(text, 0, text.length());
	bw.newLine();
	bw.flush();
      }
    } catch (IOException ioe) {
      System.out.println("Error writing file ...... ");
    }
  }
  
  public void initializeLatencyArray(int size) {
    clientProtocolHandler.initializeLatencyArray(size);
  }
  
  public void loopThroughMessageSizes(String reportFileName) {
    clientProtocolHandler.setLoopThroughMessageSizes(reportFileName);
    resultsDirectory = reportFileName;
    File resultsDir = new File(reportFileName);
    if (!resultsDir.exists()) 
      resultsDir.mkdirs();
    if (resultsDir.exists()) {
      System.out.println("Will be gathering results in directory " +
			 reportFileName);
    } else {
      System.out.println("Error creating directory " + reportFileName);
    }
  }
  
  public void startTestingProcess(String serverName, int portNum) {
    numOfMessages = 50;
    long waitForGatheringResults;
    for (int i =0; i<matchRates.length; i++) {
      setMatchRate(matchRates[i]);
      for (int j=0; j < virtualPublishRates.length; j++) {
	pubRate = virtualPublishRates[j];
	messagesAtATime = numMessageCompensation[j];
	System.out.println("Publish Rate = " + publishRates[j]);
	for (int k=0; k < messageSizes.length; k++) {
	  System.out.println("Message Size =" + messageSizes[k]);
	  eventPropagationRequest(serverName +":"+ portNum, 
				  "t", messageSizes[k]);
	  try {
	    waitForGatheringResults = (110000/matchRates[i]) + 3500;
	    Thread.sleep(waitForGatheringResults);
	  } catch (InterruptedException ie) {
	    ie.printStackTrace();
	  }
	} /** message sizes */
	try {
	  Thread.sleep(5000);
	} catch (InterruptedException ie) {
	   ie.printStackTrace();
	}
      } /** virtual pub rates */
      try {
	Thread.sleep(1000);
      } catch (InterruptedException ie) {
	ie.printStackTrace();
      }
    } /** match-rates */

  } /*end-method */
  
  public void initializeReportFile(String fileName) {
    try {
      bw2 = new BufferedWriter(new FileWriter(new File(fileName + ".txt")));
      bw4 = new BufferedWriter(new FileWriter(new File(fileName + 
						       "Latency.txt"))); 
      System.out.println("Initialized Report File......");
    } catch (IOException ioe) {
      System.out.println("Error writing file ...... ");
    } 
  }
  
  public void writeToReportFile(String reportLine) {
    try {
      bw2.write(reportLine, 0, reportLine.length());
      bw2.newLine();
      bw2.flush();
    } catch (IOException ioe) {
      System.out.println("Error writing file ...... ");
    } 
  }
  
  public void writeLatenciesToFile(long[] acceptedLatencies) {
    try {
      for (int i=0; i <acceptedLatencies.length; i++) {
	String latencyString = acceptedLatencies[i] + "";
	bw4.write(latencyString, 0, latencyString.length());
	bw4.newLine();
      }
      
      bw4.flush();
    } catch (IOException ioe) {
      System.out.println("Error writing file ...... ");
    } 
  }
  
  public void writePublishRate(float publishRate) {
    try {
      bw3 = new BufferedWriter(new FileWriter(new File("PublishRate.txt")));
      String publishString = publishRate + " ";
      bw3.write(publishString, 0, publishString.length());
      bw3.newLine();
      bw3.flush();
    } catch (IOException ioe) {
      System.out.println("Error writing file ...... ");
    }
  }
  
  public float readPublishRate() {
    try {
      BufferedReader br = 
	new BufferedReader(new FileReader(new File("PublishRate.txt")));
      String publishString = br.readLine();
      if (publishString == null) {
	br.close();
	return pubValueFromFile;
      }
      StringTokenizer tok = new StringTokenizer(publishString, ".");
      if (tok.countTokens() ==2 )
	pubValueFromFile = Float.valueOf(publishString).floatValue();
      else 
	System.out.println("Occasion when publish rate string" + 
			   " was the empty string");
      br.close();
    } catch (IOException ioe) {
      System.out.println("Error writing file ...... ");
    }
    return pubValueFromFile;
  }
  
  public static void main(String[] args) {
    if (args.length == 0) {
      System.out.println("Usage: java cgl.narada.node.ClientNode portNum " +
			 " [default:TCP] ");
      System.exit(0);
    }
    
    
    int portNum = Integer.parseInt(args[0]);
    ClientNode clientNode = new ClientNode();
    Properties props = new Properties();
    props.put("TCPServerPort", Integer.toString(portNum));
    props.put("truststore", "D:/Deliverables/SSLTunnel/keys/truststore");
    props.put("keystore", "D:/Deliverables/SSLTunnel/keys/keystore");
    props.put("truststorePassword", "cooldude");
    props.put("keystorePassword", "cooldude");
    props.put("username", "test1");
    props.put("password", "test1");
    props.put("https.proxyHost", "nile");
    props.put("https.proxyPort", "8080");
    props.put("secure", "false");
    props.put("listenerport", "443");
    
    clientNode.setUpNodeNew(props, "tcp");
    //clientNode.setUpNode(portNum, Protocol.TCP);

    BufferedReader r = new BufferedReader(new InputStreamReader(System.in));
    String cmd;
    StringTokenizer tok;
    System.out.println("\n" + "Type h for help and Usage indicators" + "\n");
    while (true) {
      try {
	cmd = r.readLine();
	if (cmd.equals("q")) break;
	if (cmd == null) continue;
	if(cmd.startsWith("h")) {
	  System.out.println("Connect                  -> c server " + 
			     "portNum t/u/m");
	  System.out.println("Node Address Request     -> na server:portNum");
	  System.out.println("Profile Propagate        -> pp server:portNum "+
			     "tag1=value1,tag2=value2");
	  System.out.println("Set publish rates        -> r pubRate matchRate numOfMessages");
	  System.out.println("Publish Event            -> e server:portNum "+
			     "tag1=value1,tag2=value2 minSizeOfEvent");
	  System.out.println("Publish Persistent Event -> e server:portNum "+
			     "tag1=value1,tag2=value2 minSizeOfEvent");
	}
	if (cmd.startsWith("c")) {
	  tok = new StringTokenizer(cmd, " ");
	  if (tok.countTokens() != 4) {
	    System.out.println("Usage -> c server portNum t/u/m");
	    continue;
	  }
	  tok.nextToken();
	  String host = tok.nextToken();
	  int port = Integer.parseInt(tok.nextToken());
	  byte transport= Protocol.TCP;
	  String medium= tok.nextToken();

	  if (medium.equals("t")) transport = Protocol.TCP;
	  else if (medium.equals("u"))  transport = Protocol.UDP;
	  else if (medium.equals("m"))  transport = Protocol.MULTICAST;
	  else if (medium.equals("h")) transport = Protocol.HTTP;
	  clientNode.connectTo(host, port, transport);
	} 
	
	if (cmd.startsWith("na")) {
	  tok = new StringTokenizer(cmd, " ");
	  if (tok.countTokens() != 2) {
	    System.out.println("Usage -> na server:portNum");
	    continue;
	  }
	  tok.nextToken();
	  String server = tok.nextToken();
	  clientNode.nodeAddressRequest(server);
	}
	if (cmd.startsWith("i")) {
	  tok = new StringTokenizer(cmd, " ");
	   if (tok.countTokens() != 2) {
	    System.out.println("Usage -> i latencyArraySize");
	    continue;
	   }
	   tok.nextToken();
	 clientNode.initializeLatencyArray(Integer.parseInt(tok.nextToken()));
	}

	if (cmd.startsWith("g")) {
	  tok = new StringTokenizer(cmd, " ");
	  if (tok.countTokens() != 4) {
	    //System.out.println("Usage -> g pubRate messageSize");
	    System.out.println("Usage -> g pubRate matchRate messageSize");
	    continue;
	  }
	  tok.nextToken();
	  int pubRate = Integer.parseInt(tok.nextToken());
	  int matchRate = Integer.parseInt(tok.nextToken());
	  int messageSize = Integer.parseInt(tok.nextToken());
	  clientNode.gatherStats(pubRate, matchRate, messageSize);
	}
	
	if (cmd.startsWith("l")) {
	  tok = new StringTokenizer(cmd, " ");
	  if (tok.countTokens() != 2) {
	    System.out.println("Usage -> l reportFileName");
	    continue;
	  }
	  tok.nextToken();
	  clientNode.loopThroughMessageSizes(tok.nextToken());
	}

	if (cmd.startsWith("pRate")) {
	  tok = new StringTokenizer(cmd, " ");
	  if (tok.countTokens() != 2) {
	    System.out.println("Usage -> pRate value ");
	    continue;
	  }
	  int pubRate = Integer.parseInt(tok.nextToken());
	  clientNode.setPublishRate(pubRate);
	}
	
	if (cmd.startsWith("t")) {
	  tok = new StringTokenizer(cmd, " ");
	  if (tok.countTokens() != 3) {
	    System.out.println("Usage -> t server portNum ");
	    continue;
	  }
	  tok.nextToken();
	  String serverName = tok.nextToken();
	  int _portNum = Integer.parseInt(tok.nextToken());
	  clientNode.startTestingProcess(serverName, _portNum);
	}
	if (cmd.startsWith("pp")) {
	  tok = new StringTokenizer(cmd, " ");
	  if (tok.countTokens() != 3) {
	    System.out.println("Usage -> pp server:portNum "+
			       "tag1=value1,tag2=value2");
	    continue;
	  }
	  tok.nextToken();
	  String server = tok.nextToken();
	  String subscription = tok.nextToken();
	  clientNode.profilePropagationRequest(server, subscription);
	}

	if (cmd.startsWith("e")) {
	  tok = new StringTokenizer(cmd, " ");
	  if (tok.countTokens() != 4) {
	    System.out.println("Usage -> e server:portNum " +
			       "tag1=value1,tag2=value2 minSizeOfEvent");
	    continue;
	  }
	  tok.nextToken();
	  String server = tok.nextToken();
	  String event = tok.nextToken();
	  int minSize = Integer.parseInt(tok.nextToken());
	  clientNode.eventPropagationRequest(server, event, minSize);
	  
	}

	if (cmd.startsWith("pe")) {
	  tok = new StringTokenizer(cmd, " ");
	  if (tok.countTokens() != 4) {
	    System.out.println("Usage -> pe server:portNum " +
			       "tag1=value1,tag2=value2 minSizeOfEvent");
	    continue;
	  }
	  tok.nextToken();
	  String server = tok.nextToken();
	  String event = tok.nextToken();
	  int minSize = Integer.parseInt(tok.nextToken());
	  clientNode.persistentEventPropagationRequest(server, event, minSize);
	}

	if (cmd.startsWith("r")) {
	  tok = new StringTokenizer(cmd, " ");
	  if (tok.countTokens() != 5) {
	    System.out.println("Usage -> r pubRate matchRate numOfMessages messagesAtATime");
	    continue;
	  }
	  tok.nextToken();
	  int pubRate = Integer.parseInt(tok.nextToken());
	  int matchRate =  Integer.parseInt(tok.nextToken());
	  int numOfMessages = Integer.parseInt(tok.nextToken());
	  int messagesAtATime = Integer.parseInt(tok.nextToken());
	  clientNode.setRates(pubRate, numOfMessages, messagesAtATime);
	  clientNode.setMatchRate(matchRate);
	}
	
	if (cmd.startsWith("send")) {
	  tok = new StringTokenizer(cmd, " ");
	  tok.nextToken();
	  String client = tok.nextToken();
	  byte[] data = tok.nextToken().getBytes();
	  //protocolHandler.sendTo(client, data);
	}
      }catch (IOException ioe) {
	ioe.printStackTrace();
      }
    }
    System.out.println("Program Exiting Now");
  }
}
